package simpletestbed;

import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.FileReader;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import simpletestbed.core.SimpleTestBedClient;
import simpletestbed.core.TestServerCaller;

public class RemoteTestBed extends TestBed {
	
	protected List<String> remoteServers = new ArrayList<String>();
	
	protected List<Integer> remoteServerCapacities = new ArrayList<Integer>();
	
	protected String localBinPath = "bin";
	
	protected double[] avgRuntimes = null;
	
	protected double[] maxMemUsage = null;
	
	public RemoteTestBed(String configFileName) throws IOException {
		FileReader freader = new FileReader(configFileName);
		BufferedReader reader = new BufferedReader(freader);
		String line;
		while((line=reader.readLine())!= null) {
			if (line.length() == 0) {
				continue;
			}
			String[] tokens = line.split(" ");
			String serverHost = tokens[0];
			this.remoteServers.add(serverHost);
			if (tokens.length >= 2) {
				int capacity = Integer.parseInt(tokens[1]);
				this.remoteServerCapacities.add(capacity);
			}
		}
		reader.close();
		freader.close();
	}
	
	public RemoteTestBed(String configFileName, String localBinPath) throws IOException {
		this(configFileName);
		this.localBinPath = localBinPath;
	}
	
	public int getServerCapacity(int serverIndex) {
		return this.remoteServerCapacities.get(serverIndex);
	}

	@Override
	public double[] getRuntimes() {
		return this.avgRuntimes;
	}
	
	@Override
	public double[] getMaxMemUsages() {
		return this.maxMemUsage;
	}

	@Override
	public TestResult[] execute() {
		try {
			int numMachines = this.remoteServers.size();
			if (numMachines == 0) {
				throw new Error("No remote server configured !");
			}
			
			// Allocate the test cases for machines
			int numCasePerMachine;
			if (this.testCases.size() <= numMachines) {
				numCasePerMachine = 1;
				numMachines = this.testCases.size();
			}
			else {
				numCasePerMachine = (int)(((double)this.testCases.size()) / ((double)numMachines)+0.5);
			}
			RemoteMonitor[] monitors = new RemoteMonitor[numMachines];
			int startCaseIdx = -1;
			int endCaseIdx = -1;
			for (int machineId=0; machineId<numMachines; machineId++) {
				startCaseIdx = numCasePerMachine*machineId;
				if (machineId == numMachines - 1) { // Last machine
					endCaseIdx = this.testCases.size();
				}
				else { // Not last machine
					endCaseIdx = startCaseIdx + numCasePerMachine;
				}
				List<TestCase> subCases = new ArrayList<TestCase>();
				for (int i=startCaseIdx; i<endCaseIdx; i++) {
					subCases.add(this.testCases.get(i));
				}
				String serverHost = this.remoteServers.get(machineId);
				monitors[machineId] = new RemoteMonitor(serverHost, machineId, subCases,
						this.localBinPath, this.checkMemUsage, this.threadTimeOut);
			}
			
			// Start to run all test cases
			for (int machineId=0; machineId<numMachines; machineId++) {
				monitors[machineId].start();
			}
			
			// Wait for results
			TestResult[] averageResults = new TestResult[this.testCases.size()];
			this.avgRuntimes = new double[this.testCases.size()];
			this.maxMemUsage = new double[numMachines];
			startCaseIdx = 0;
			for (int machineId=0; machineId<numMachines; machineId++) {	
				// Wait for the thread ending
				monitors[machineId].join();
				// Copy the result scores
				int numSubCases = monitors[machineId].getTestCases().size();
				TestResult[] subResults = monitors[machineId].getResult();
				double[] subRuntimes = monitors[machineId].getRuntime();
				this.maxMemUsage[machineId] = monitors[machineId].getMaxMemUsage();
				System.arraycopy(subResults, 0, averageResults, startCaseIdx, numSubCases);
				System.arraycopy(subRuntimes, 0, this.avgRuntimes, startCaseIdx, numSubCases);
				startCaseIdx += numSubCases;
			}
			return averageResults;
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			return null;
		}		
	}
	
	/**
	 * The thread for sending sub test cases to remote servers
	 * @author Liang
	 *
	 */
	static class RemoteMonitor extends Thread {
		
		String serverName;
		int    serverId;
		List<TestCase> testCases;
		TestResult[] results;
		double[] runtimes;
		boolean  checkMemUsage;
		double   maxMemUsage;
		String   localBinPath;
		long     threadTimeOut; // 10 minutes
		
		public RemoteMonitor(String serverName, int serverId, List<TestCase> testCases,
				String localBinPath, boolean checkMemUsage, long timeout)  {
			this.serverName = serverName;
			this.serverId = serverId;
			this.testCases = testCases;
			this.localBinPath = localBinPath;
			this.threadTimeOut = timeout;
			this.checkMemUsage = checkMemUsage;
		}
		
		public RemoteMonitor(String serverName, int serverId, List<TestCase> testCases, 
				boolean checkMemUsage, long timeout)  {
			this(serverName, serverId, testCases, null, checkMemUsage, timeout);
		}
		
		public String getServerName() {
			return this.serverName;
		}
		
		public int getServerId() {
			return this.serverId;
		}
		
		public TestResult[] getResult() {
			return this.results;
		}
		
		public double[] getRuntime() {
			return this.runtimes;
		}
		
		public double getMaxMemUsage() {
			return this.maxMemUsage;
		}
		
		public List<TestCase> getTestCases() {
			return this.testCases;
		}

		@Override
		public void run() {
			// TODO Auto-generated method stub
			try {
				// Convert the test cases into argument byte buffer
				byte[] arg = null;
				ByteArrayOutputStream baos = new ByteArrayOutputStream();
				DataOutputStream dos = new DataOutputStream(baos);
				dos.writeInt(testCases.size());
				dos.flush();
				for (int i=0; i<testCases.size(); i++) {
					ObjectOutputStream objStream = new ObjectOutputStream(dos);
					objStream.writeObject(testCases.get(i));
					objStream.flush();
				}
				dos.flush();
				arg = baos.toByteArray();	
				
				// Invoke SimpleTestBed client to send command
				SimpleTestBedClient client = new SimpleTestBedClient(serverName);
				if (this.localBinPath != null) {
					// Send java classes
					client.execSendClasses("clientcp"+this.serverId, this.localBinPath);
				}
				// Invoke TestServerCaller's function remotely
				String className = TestServerCaller.class.getName();
				String methodName = TestServerCaller.CALLER_FUNC_NAME;
				byte[] resultBuf = client.execRun(className, methodName, this.checkMemUsage, 
						this.threadTimeOut, arg);
				// Convert the result into average scores
				ByteArrayInputStream bais = new ByteArrayInputStream(resultBuf);
				ObjectInputStream ois = new ObjectInputStream(bais);
				this.results = new TestResult[this.testCases.size()];
				this.runtimes = new double[this.testCases.size()];
				for (int i=0; i<this.testCases.size(); i++) {
					this.results[i] = (TestResult)ois.readObject();
					this.runtimes[i] = ois.readDouble();
				}
				this.maxMemUsage  = ois.readDouble();
				ois.close();
				bais.close();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		
	}

}
